package com.hsc.www.redis.jedisLettuce.jedis;

import com.alibaba.fastjson.JSONObject;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.RedisURI;
import io.lettuce.core.TransactionResult;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.async.RedisStringAsyncCommands;
import io.lettuce.core.api.reactive.RedisStringReactiveCommands;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import io.lettuce.core.pubsub.RedisPubSubListener;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
import org.junit.Test;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class LettuceClient {

    /**
     * 使用jedis：当多线程使用同一个连接时，是线程不安全的。所以要使用连接池，为每个jedis实例分配一个连接。
     * 使用Lettuce：当多线程使用同一连接实例时，是线程安全的。
     */
    @Test
    public void test1() {
        // client
        RedisClient client = RedisClient.create("redis://106.12.180.184:6380");
        // connection, 线程安全的长连接，连接丢失时会自动重连，直到调用 close 关闭连接。
        StatefulRedisConnection<String, String> connection = client.connect();

        // sync, 默认超时时间为 60s.
        RedisCommands<String, String> redisCommands = connection.sync();
        redisCommands.set("name", "老五");
        String value = redisCommands.get("name");
        System.out.println("姓名：" + value);

        // close connection
        connection.close();
        // shutdown
        client.shutdown();
    }

    @Test
    public void test2() {
        //创建RedisURI的两种方式
        RedisURI redisURI1 = RedisURI.create("redis://106.12.180.184:6380");
        RedisURI redisURI2 = RedisURI.Builder.redis("106.12.180.184").withPort(6381).withDatabase(0).build();
        List<RedisURI> list = Arrays.asList(redisURI1, redisURI2);

        RedisClusterClient client = RedisClusterClient.create(list);
        //超时时间
        client.setDefaultTimeout(Duration.ofSeconds(20));

        StatefulRedisClusterConnection<String, String> connect = client.connect();

        /* 同步执行的命令 */
        RedisAdvancedClusterCommands<String, String> commands = connect.sync();
        commands.set("name", "老五");
        String value = commands.get("name");
        System.out.println("姓名：" + value);

        /*  异步执行的命令  */
//      RedisAdvancedClusterAsyncCommands<String, String> commands= connect.async();
//      RedisFuture<String> future = commands.get("test2");
//      try {
//          String str = future.get();
//          System.out.println(str);
//      } catch (InterruptedException e) {
//          e.printStackTrace();
//      } catch (ExecutionException e) {
//          e.printStackTrace();
//      }

        connect.close();
        client.shutdown();
    }

    //    Synchronous methods will throw a RedisCommandExecutionException in case Redis responds with an error.
// Asynchronous connections do not throw exceptions when Redis responds with an error.
    @Test
    public void test3() throws Exception {
        // client
        RedisClient client = RedisClient.create("redis://106.12.180.184:6380");
        StatefulRedisConnection<String, String> connection = client.connect();

        // async
        RedisStringAsyncCommands<String, String> asyncCommands = connection.async();

        asyncCommands.set("name", "老五");
        RedisFuture<String> future = asyncCommands.get("name");
        future.thenAccept((str) -> {
            System.out.println("姓名：" + str);
        });

        System.out.println("END");
        Thread.sleep(10 * 1000);
        connection.close();
        client.shutdown();
    }

    @Test
    public void test4() throws Exception {

        // client
        RedisClient client = RedisClient.create("redis://106.12.180.184:6380");

        // connect
        StatefulRedisConnection<String, String> connection = client.connect();

        // reactive
        RedisStringReactiveCommands<String, String> reactiveCommands = connection.reactive();
        reactiveCommands.set("name", "老五");
        reactiveCommands.get("name").subscribe((str) -> {
            System.out.println("姓名：" + str);
        });

        System.out.println("END");
        Thread.sleep(10 * 1000);
        connection.close();
        client.shutdown();
    }

    @Test
    public void test5() throws Exception {

        ExecutorService threadPool = Executors.newFixedThreadPool(2);
        //发布Pub
        threadPool.submit(() -> {
            RedisClient client = RedisClient.create("redis://106.12.180.184:6380");
            StatefulRedisPubSubConnection<String, String> pubSubConnection = client.connectPubSub();
            RedisPubSubCommands<String, String> pubSubCommands = pubSubConnection.sync();
            while (true) {
                pubSubCommands.publish("CCTV1", "新文联播" + System.currentTimeMillis());
                pubSubCommands.publish("CBA", "篮球" + System.currentTimeMillis());
                pubSubCommands.publish("BCC", "动物世界" + System.currentTimeMillis());
                Thread.sleep(2000);
            }
        });

        //订阅Sub
        threadPool.submit(() -> {
            RedisClient client = RedisClient.create("redis://106.12.180.184:6380");
            StatefulRedisPubSubConnection<String, String> pubSubConnection = client.connectPubSub();
            pubSubConnection.addListener(new MyRedisPubSubListener());
            RedisPubSubCommands<String, String> pubSubCommands = pubSubConnection.sync();
            pubSubCommands.subscribe("BCC");
            pubSubCommands.psubscribe("CCTV*");
        });

        while (true) {
            Thread.sleep(1000);
        }
    }

    @Test
    public void test6() throws Exception {
        RedisClient client = RedisClient.create("redis://106.12.180.184:6380");
        StatefulRedisConnection<String, String> connect = client.connect();

        RedisAsyncCommands<String, String> commands = connect.async();

        commands.multi();
        commands.set("name", "老五");
        commands.set("age", "60");
        commands.get("name");
        commands.get("age");

        RedisFuture<TransactionResult> exec = commands.exec();
        TransactionResult result = exec.get();

        for (int i = 0;i  < result.size(); i++) {
            System.out.println(result.get(i).toString());
        }
        while (true) {
            Thread.sleep(1000);
        }
    }




//    RedisCommandFactory factory = new RedisCommandFactory(client.connect());

}


class MyRedisPubSubListener implements RedisPubSubListener<String, String> {

    //===============普通频道====================
    @Override
    public void subscribed(String channel, long count) {
        System.out.println(toJson(new HashMap() {{
            put("methodName", "订阅频道");
            put("channel", channel);
            put("count", count);
        }}));
    }

    @Override
    public void message(String channel, String message) {
        System.out.println(toJson(new HashMap() {{
            put("methodName", "从频道订阅中接收消息");
            put("channel", channel);
            put("message", message);
        }}));
    }

    @Override
    public void unsubscribed(String channel, long count) {
        System.out.println(toJson(new HashMap() {{
            put("methodName", "取消订阅频道");
            put("channel", channel);
            put("count", count);
        }}));
    }

    //================给定模式频道===================
    @Override
    public void psubscribed(String pattern, long count) {
        System.out.println(toJson(new HashMap() {{
            put("methodName", "订阅给定模式频道");
            put("pattern", pattern);
            put("count", count);
        }}));
    }

    @Override
    public void message(String pattern, String channel, String message) {
        System.out.println(toJson(new HashMap() {{
            put("methodName", "从给定模式频道订阅中接收消息");
            put("pattern", pattern);
            put("channel", channel);
            put("message", message);
        }}));
    }

    @Override
    public void punsubscribed(String pattern, long count) {
        System.out.println(toJson(new HashMap() {{
            put("methodName", "取消订阅给定模式频道");
            put("pattern", pattern);
            put("count", count);
        }}));
    }

    private String toJson(Object obj) {
        return JSONObject.toJSONString(obj);
    }
}








